package com.spring.demo.webflux.demo.test;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author wangzhilong
 * @date 2021/3/15
 */
public class FluxDemo {

    public static void main(String[] args) throws InterruptedException {
//        generateDemo();

        //作为一个更高级的创建 Flux 的方式， create 方法的生成方式既可以是同步， 也可以是异步的，并且还可以每次发出多个元素。
        //create 有个好处就是可以将现有的 API 转为响应式，比如监听器的异步方法。
        Flux.create( sink -> {
        });

        Mono.fromCallable(() -> {
            Thread.sleep(1000);
            System.out.println("远程方法调用完成："+ Thread.currentThread());
            return "ok";
        })
        //使用 Schedulers.elastic() 确保对每一个订阅来说运行在一个专门的线程上
        .subscribeOn(Schedulers.elastic())
        .subscribe(System.out::println);
        Thread.sleep(1000);

        Flux.range(1, 10000)
                //创建等同于 CPU 个数的线程（最小为4）。
                .publishOn(Schedulers.parallel())
                .subscribe();

        //为了说明，这里有个读写 Context 的简单例子：
        //自从版本 3.1.0，Reactor 引入了一个类似于 ThreadLocal 的高级功能：Context。
        // 它作用于一个 Flux 或一个 Mono 上，而不是应用于一个线程（Thread）。
        String key = "message";
        Mono.just("hello")
                //对源调用 flatMap 用 Mono.subscriberContext() 方法拿到 Context。
                .flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
                //此处顺序不能再flatMap之前
                //实际上，一个 Context 是绑定到每一个链中的 Subscriber 上的。
                // 它使用 Subscription 的传播机制来让自己对每一个操作符都可见（从最后一个 subscribe 沿链向上）。
                //操作链以调用 subscriberContext(Function) 结尾，将 "World" 作为 "message" 这个 key 的 值添加到 Context 中。
                //subscriberContext 是链上的最后一个环节，但确实最先执行的（原因在于 subscription 信号 是从下游向上的）。
                .subscriberContext(ctx -> ctx.put(key, "World"))
                .subscribe(System.out::println);


        Mono.subscriberContext()
                .map(ctx -> ctx.put(key, "hello"))
                .flatMap(ctx -> Mono.subscriberContext())
                .map(ctx -> ctx.getOrDefault(key, "default"))
                .subscribe(System.out::println);

        //原因在于 Context 是与 Subscriber 关联的，而每一个操作符访问的 Context 来自其下游的 Subscriber。
        Mono.just("hello")
                .flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "Reactor"))
                .flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "World"))
                .subscribe(System.out::println);

        //最后一个有意思的传播
        Mono.just("Hello")
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.get(key))
                )
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.get(key))
                        //这个 subscriberContext 不会影响所在 flatMap 之外的任何东西。
                        .subscriberContext(ctx -> ctx.put(key, "Reactor"))
                )
                //这个 subscriberContext 会影响主序列的 Context。
                .subscriberContext(ctx -> ctx.put(key, "World"))
                .subscribe(System.out::println);

        System.out.println("-------------");

        Mono.just("unknown")
                .flatMap(s -> {
                    if (s.equalsIgnoreCase("unknown")) {
                        throw new RuntimeException("未知错误");
                    }
                    return Mono.just(s);
                })
//                .doOnError(error -> System.out.println("doOnError->" + error))
                .onErrorResume(e -> Mono.just("处理异常成功"))
                .doOnNext(s -> System.out.println("doOnNext->" + s))
                .subscribe(s -> System.out.println("subscribe->" + s));

        System.out.println("============");
        Flux.range(1, 10)
                .log()
                .take(1)
                .single()
                .subscribe();

    }

    private static void generateDemo() {
        //初始化一个可变对象作为状态变量。
        //这是一种 同步地， 逐个地 产生值的方法
        Flux.generate(
                () -> 0,
                (state, sink) -> {
                    sink.next("3 x " + state + " = " + 3 * state);
                    if (state == 10) {
                        sink.complete();
                    }
                    return state + 1;
                }
            )
                .doOnNext(System.out::println)
                .subscribe();

        Flux.generate(AtomicLong::new,
                (state, sink) -> {
                    final long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3 * i);
                    if (i == 10) {
                        sink.complete();
                    }
                    return state;
                })
                .subscribe(System.out::println);
    }

    interface  MyEventListenner<T> {

        /**
         * 一个 chunk 数据准备好的事件
         * @param chunk
         */
        void onDataChunk(List<T> chunk);

        /**
         * 处理结束的事件
         */
        void processComplete();
    }

    class MyEventProcess {
        void register(MyEventListenner listenner) {

        }
    }


}
